package com.mjf.spark.day10

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 通过DirectAPI连接Kafka数据源，获取数据
 *    手动维护offset
 */
object SparkStreaming07_DirectAPI_Hander {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")

    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 准备Kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )

    // 获取上一次消费的位置（偏移量）
    // 实际项目中，为了保证数据精准一致性，我们对数据进行消费处理后，将偏移量保存在有事务的存储中，如MySQL
    def fromOffsets(): Map[TopicAndPartition, Long] = {
      Map[TopicAndPartition, Long](
        TopicAndPartition("bigdata-mjf", 0) -> 1L,
        TopicAndPartition("bigdata-mjf", 1) -> 1L
      )
    }

    // 从指定的offset读取数据进行消费
    val kafkaDStream: InputDStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
      ssc,
      kafkaParams,
      fromOffsets,
      (m: MessageAndMetadata[String, String]) => m.message()
    )

    // 定义一个空集合来维护偏移量
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]

    // 消费完毕之后，对偏移量offset进行更新
    kafkaDStream.transform{
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    }.foreachRDD{
      rdd => {
        for (o <- offsetRanges) {
          println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
        }
      }
    }

    // 启动采集器
    ssc.start()

    // 等待采集结束之后，终止程序
    ssc.awaitTermination()

  }

}
